/*
 * REFThread.cpp
 *
 *  Created on: Apr 19, 2013
 *      Author: Lucifer
 */

#include "../pch.h"

#ifdef	_WIN32
#include <process.h>

#define	APC_ALERT	1234

#else	//	_WIN32
#include <unistd.h>	/* usleep */
#include <signal.h>
#include <pthread.h>

#ifdef	__APPLE_CC__
#define	YieldProcessor()	pthread_yield_np()
#else	//	__APPLE_CC__
#define	YieldProcessor()	pthread_yield()
#endif	//	!__APPLE_CC__

#define	SIG_ALERT	(SIGUSR1 + 1)
#define INVALID_HANDLE_VALUE ((HANDLE)(-1))

#endif	//	!_WIN32

namespace BFX
{

//////////////////////////////////////////////////////////////////////////////

/**
 * Initializes an new instance of BFX::REFCallback class.
 */
REFThreadCallback::REFThreadCallback()
{
}

/**
 * Initializes an new instance of BFX::REFRunnable class.
 * @param[in] pfnCallback The pointer of method to be invoked when this object begins running.
 */
REFThreadCallback::REFThreadCallback(ThreadCallback fnCallback)
{
	m_fnCallback = fnCallback;
}

/**
 * Destroys the BFX::REFRunnable object.
 */
REFThreadCallback::~REFThreadCallback()
{
}

/**
 * This method uses to be called in that separately time or running thread.
 */
void REFThreadCallback::Run()
{
		m_fnCallback();
}

//////////////////////////////////////////////////////////////////////////////
// ThreadStateHolder

// helper class to hold thread state.
class ThreadStateHolder
{
public:
	ThreadStateHolder(bool bNeed, LONG dwState)
		: m_bNeed(bNeed), m_lState(dwState)
	{
		BFX_ASSERT(REFThread::GetCurrentThread());
	}
	~ThreadStateHolder()
	{
		if (m_bNeed)
		{
			REFThread *pCurThread = REFThread::GetCurrentThread();
			BFX_ASSERT (pCurThread);
			Interlocked::And(&pCurThread->m_lState, ~m_lState);
		}
	}

private:
	bool m_bNeed;
	LONG m_lState;
};

//////////////////////////////////////////////////////////////////////////////
// REFThread

//
// static variables
//
bool		REFThread::s_bInitialized = false;
SpinLock	REFThread::s_syncLock;

#ifdef	_WIN32
UINT32 REFThread::s_dwTlsSlot	= 0;
#else
pthread_key_t REFThread::s_hTlsSlot = (pthread_key_t)INVALID_HANDLE_VALUE;
#endif

/**
 * Initializes a new instance of the BFX::REFThread class.
 * the virtual method BFX::REFThread::Run() to be invoked when this thread begins executing.
 */
REFThread::REFThread()
{
	Initialize();
}

/**
 * Initializes a new instance of the the BFX::REFThread class.
 * @param[in] pStart The pointer of BFX::REFCallback to be invoked when this thread begins executing.
 */
REFThread::REFThread(REFThreadCallback* pStart)
{
	BFX_REQUIRE(pStart != NULL);

	Initialize();

	m_pStart = pStart;
}

/**
 * Initializes a new instance of the the BFX::REFThread class.
 * @param[in] fnStart The functor object to be invoked when this thread begins executing.
 */
REFThread::REFThread(ThreadCallback fnStart)
{
	// BFX_REQUIRE(pfnStart != NULL);

	Initialize();

	m_pStart = new REFThreadCallback(fnStart);
}

/**
 * Destroys the BFX::REFThread object.
 */
REFThread::~REFThread()
{
	if (m_hThread == INVALID_HANDLE_VALUE)
		return;

#ifdef	_WIN32
	UINT nCurThreadId = ::GetCurrentThreadId();
	if (m_nThreadId == nCurThreadId)
#else
	pthread_t hCurThread = ::pthread_self();
	if (m_hThread == (HANDLE)hCurThread)
#endif
	{
		// destroy thread self.
#ifdef	_WIN32
		::ExitThread(0);
#else
		::pthread_exit(NULL);
#endif
		return;
	}
	{
		// terminate thread form other thread.
		Abort();
		Join();
	}
}

/**
 * Starts the thread, causes the thread state to running.
 */
void REFThread::Start()
{
#ifdef	_WIN32
	UINT32 nThreadId = 0;
	// create the new thread.
	HANDLE hThread = (HANDLE)::_beginthreadex(NULL, 0, IntermediateThreadProc, this, 0, &nThreadId);
	BFX_ENSURE1(hThread != INVALID_HANDLE_VALUE, "failed to create new thread.");

	Interlocked::ExchangePointer(&m_hThread, hThread);
	BFX_ASSERT(nThreadId != 0);
	m_nThreadId = nThreadId;
#else	//	_WIN32
	int nRetVal = 0;
	pthread_attr_t threadAttr;
	pthread_t hThread = NULL;

	nRetVal = ::pthread_attr_init(&threadAttr);
	BFX_ENSURE1(nRetVal == 0, "couldn't initialize thread attributes.");

	nRetVal = ::pthread_create(&hThread, &threadAttr, IntermediateThreadProc, this);
	::pthread_attr_destroy(&threadAttr);
	BFX_ENSURE1(nRetVal == 0, "failed to creates new thread.");

	// thread ID is set to the the pthread ID (pthread_self)
	Interlocked::ExchangePointer((void**)&m_hThread, (void*)hThread);
#endif	//	!_WIN32

	// NOTE: need to blocks the caller thread until target is really running???
	// synchronize with HasStarted.
	while (m_lState & TS_Unstarted)
	{
		YieldProcessor();
	}
}

/**
 * Raises a BFX::ThreadInterruptException in the thread on which it is invoked, to interrupts a thread that is in
 * the Wait Sleep Join thread state.
 */
void REFThread::Interrupt()
{
	Interlocked::Or(&m_lUserInterrupt, TI_Interrupt);
	if (m_lState & TS_Interruptible)
	{
		Alert();
	}
}

#define	ABORT_POLL_TIMEOUT	10
#ifdef	_DEBUG
#define	ABORT_FAIL_TIMEOUT	40000
#endif	// _DEBUG

/**
 * Raises a BFX::ThreadAbortException in the thread on which it is invoked, to begin the process of terminating the
 * thread. Calling this method usually terminates the thread.
 */
void REFThread::Abort()
{
	if (m_hThread == INVALID_HANDLE_VALUE)
		return;	// Can not abort a detached thread.

	// if aborting self
	REFThread *pCurThread = GetCurrentThread();
	if (this == pCurThread)
	{
		// abort point: 1
		SetAbortInitiated();

		// I hate C++ exception, but we have to do it.
		ThrowAlertable(true);
	}

	BFX_ASSERT(this != pCurThread);
#ifdef	_DEBUG
	UINT32	dwElapsedTime = 0;
#endif
	for (;;)
	{
		// we own TS lock.  the state of m_hThread can not be changed.
		if (m_lState & TS_Unstarted)
		{
			// this thread is not yet started.
			// abort point: 2
			return;
		}

		if (m_hThread == INVALID_HANDLE_VALUE && (m_lState & TS_Unstarted) == 0)
		{
			// the thread is going to die or is already dead.
			// unmark thread for abort
			Interlocked::And(&m_lState, ~TS_AbortInitiated);
			Interlocked::Exchange(&m_lUserInterrupt, 0);

			// abort point: 3
			return;
		}

		// if a thread is Dead or Detached, abort is a NOP.
		if (m_lState & (TS_Unstarted | TS_Dead))
		{
			// unmark thread for abort
			Interlocked::And(&m_lState, ~TS_AbortInitiated);
			Interlocked::Exchange(&m_lUserInterrupt, 0);

			// abort point: 4
			return;
		}

		// TODO: thread abort needs to walk stack to decide if thread abort can proceed.

		// if the thread is in sleep, wait, or join interrupt it
		// however, we do NOT want to interrupt if the thread is already processing an exception
		// abort point: 5"
		if (m_lState & TS_Interruptible)
		{
			Interlocked::Or(&m_lUserInterrupt, TI_Abort);	// if the user wakes up because of this, it will read the
			                                                        // abort requested bit and initiate the abort
			Alert();
		}

        // don't do a Sleep().  It's possible that the thread we are trying to abort is
		// stuck in external code trying to get into the apartment that we are supposed
		// to be pumping!  Instead, ping the current thread's handle. obviously this
		// will time out, but it will pump if we need it to.
		if (pCurThread)
		{
			pCurThread->Join(ABORT_POLL_TIMEOUT);
		}
		else
		{
#ifdef	_WIN32
			::SleepEx(ABORT_POLL_TIMEOUT, FALSE);
#else
			::usleep(ABORT_POLL_TIMEOUT * 1000);
#endif
		}

#ifdef	_DEBUG
		dwElapsedTime += ABORT_POLL_TIMEOUT;
		if (dwElapsedTime >= ABORT_FAIL_TIMEOUT)
		{
			BFX_ASSERT1(dwElapsedTime < ABORT_FAIL_TIMEOUT, "abort timedout.");
		}
#endif
		// abort point: 6
	}	// !for(;;)
}

/*
 * The basic mechanism to send thread immediate notifications. Currently uses by interrupts and aborts the target
 * thread by other threads.
 * @see BFX::REFThread::Interrupt(), BFX::REFThread::Abort().
 */
void REFThread::Alert()
{
	HANDLE hThread = m_hThread;
	if (hThread != INVALID_HANDLE_VALUE && (m_lState & TS_Interruptible))
	{
#ifdef	_WIN32
		DWORD dwRetVal = ::QueueUserAPC(UserInterruptAPC, hThread, APC_ALERT);
		BFX_ASSERT(dwRetVal);
#else
		int nRetVal = ::pthread_kill((pthread_t)hThread, SIG_ALERT);
		BFX_ASSERT(nRetVal == 0);
#endif
	}
}

/**
 * Blocks the calling thread until a thread terminates or the specified time elapses,
 * @param nMilliseconds The number of milliseconds to wait for the thread to terminate. The default -1 indicates an
 * 		infinite timeout period.
 * @return True if the thread has terminated; false if the thread has not terminated after the amount of time
 * 		specified by the nMilliseconds has elapsed.
 */
bool REFThread::Join(INT32 nMilliseconds)
{
	BFX_REQUIRE((nMilliseconds >= 0) || (nMilliseconds == (INT32)INFINITE));

	if (m_hThread == INVALID_HANDLE_VALUE)
		return true;	// Can not join a detached thread.

	if (m_lState & (TS_Unstarted | TS_Dead))
		return true;	// The thread already dead or not start yet.

#ifdef	_WIN32
	return (WAIT_OBJECT_0 == ::WaitForSingleObject(m_hThread, (DWORD)nMilliseconds));
#else
	// NOTE: Most POSIX system not provide timeout join mechanism, so we uses a sync-block to instead.
	return m_pThreadEvent->Wait(nMilliseconds);
#endif
}

/**
 * Gets a unique identifier for the current thread.
 * @return An integer that represents a unique identifier for this thread.
 */
long REFThread::GetThreadId() const
{
#ifdef	_WIN32
	return m_nThreadId;	// NOTE: in windows system, the thread handle is not unique.
#else
	return (long) m_hThread;
#endif
}

/**
 * Suspends the current thread for a specified time.
 * @param nMilliseconds The number of milliseconds for which the thread is blocked. Default zero (0) to indicate
 * 		that this thread should be suspended to allow other waiting threads to execute; Specify Infinite to block
 * 		the thread indefinitely.
 */
void REFThread::Sleep(INT32 nMilliseconds)
{
	BFX_REQUIRE(nMilliseconds >= 0);

	// Get our thread object from TLS.
	REFThread *pCurThread = GetCurrentThread();
	bool bAlertable = true;

	if (pCurThread && bAlertable)
	{
		// A word about ordering for Interrupt.  If someone tries to interrupt a thread
		// that's in the interruptible state, we queue an APC.  But if they try to interrupt
		// a thread that's not in the interruptible state, we just record that fact.  So
		// we have to set TS_Interruptible before we test to see whether someone wants to
		// interrupt us or else we have a race condition that causes us to skip the APC.
		Interlocked::Or(&pCurThread->m_lState, TS_Interruptible);
		if (pCurThread->m_lUserInterrupt)
		{
			pCurThread->HandleThreadInterrupt();
		}

		ThreadStateHolder tsh(bAlertable, TS_Interruptible | TS_Interrupted);
		Interlocked::And(&pCurThread->m_lState, ~TS_Interrupted);

#ifdef	_WIN32
		DWORD dwRetCode = ::SleepEx(nMilliseconds, bAlertable);
		if (dwRetCode == WAIT_IO_COMPLETION)
#else
		INT32 dwRetCode = ::usleep(nMilliseconds * 1000);
		if (dwRetCode != 0)	// (errno == EINVAL) Interrupted by a signal; see signal(7).
#endif
		{
			BFX_ASSERT (bAlertable);
			// We could be woken by some spurious signal or to interrupt us.
			// In the latter case the TS_Interrupted bit will be set
			// in the thread state bits.
			// TODO: Otherwise we just go back to sleep again.
			if (pCurThread->m_lState & TS_Interrupted)
			{
				pCurThread->HandleThreadInterrupt();
			}
		}
	}
	else
	{
#ifdef	_WIN32
		::SleepEx(nMilliseconds, bAlertable);
#else
		::usleep(nMilliseconds * 1000);
#endif
	}
}

/**
 * Gets the currently running thread.
 * @return A BFX::REFThread object that is the representation of the currently running thread. Or NULL if the
 * 		current thread not created by BFX::REFThread class.
 */
REFThread* REFThread::GetCurrentThread()
{
	StaticInitialize();

	REFThread* pCurThread = NULL;

#ifdef	_WIN32
	if (s_dwTlsSlot != 0)
	{
		pCurThread = (REFThread*)::TlsGetValue(s_dwTlsSlot);

		BFX_ASSERT(pCurThread == NULL || pCurThread->m_nThreadId == ::GetCurrentThreadId());
	}
#else
	if (s_hTlsSlot != (pthread_key_t)INVALID_HANDLE_VALUE)
	{
		// fetch it hard way
		pCurThread = (REFThread*) ::pthread_getspecific(s_hTlsSlot);

		BFX_ASSERT(pCurThread == NULL || pCurThread->m_hThread == (HANDLE)::pthread_self());
	}
#endif
	return pCurThread;
}

/**
 * Causes a thread to wait the number of times defined by the iterations parameter.
 * @param nIterations A number that defines how long a thread is to wait. Default value is 1.
 *
 */
void REFThread::SpinWait(int nIterations)
{
	for (int i = 0; i < nIterations; i ++)
	{
		YieldProcessor();
	}
}

/**
 * Gets the number of logical processors.
 * @return The number of logical processors.
 */
int REFThread::GetNumberOfProcessors()
{
	static int s_nNumberOfProcessors = 0;
	if (s_nNumberOfProcessors == 0)
	{
#ifdef	_WIN32
		SYSTEM_INFO si;
		::GetSystemInfo(&si);
		s_nNumberOfProcessors = (int) si.dwNumberOfProcessors;
#else
		s_nNumberOfProcessors = sysconf(_SC_NPROCESSORS_CONF);
#endif
	}
	return s_nNumberOfProcessors;
}

/*
 * Initializes thread static / instance resources, such as thread slot; alert signals...etc.
 * NOTE: There is no necessary to call this method manually, the framework will call it when first time uses it.
 * @return True if successfully to initializes, otherwise false.
 */
void REFThread::StaticInitialize()
{
	if (!s_bInitialized)
	{
		SpinLockHolder holder(&s_syncLock);
		if (!s_bInitialized)
		{
			// Initialize thread object TLS key.
#ifdef	_WIN32
			BFX_ASSERT(s_dwTlsSlot == 0);
			s_dwTlsSlot = ::TlsAlloc();
#else
			BFX_ASSERT(s_hTlsSlot == (pthread_key_t)INVALID_HANDLE_VALUE);
			int nRetVal = ::pthread_key_create(&s_hTlsSlot, NULL);
			BFX_ASSERT1(nRetVal == 0, "pthread_key_create() failed.");

			// Install the signal handler for thread alert mechanism.
			struct sigaction sa;
			sa.sa_handler = REFThread::SignalInterruptProc;
			sigemptyset(&sa.sa_mask);
			sa.sa_flags = SA_RESTART;
			nRetVal = ::sigaction(SIG_ALERT, &sa, NULL);
			BFX_ASSERT1(nRetVal != -1, "sigaction() failed");
#endif
			s_bInitialized = true;
		}
	}
}

void REFThread::Initialize()
{
	StaticInitialize();

	m_lState = TS_Unstarted;
	m_lUserInterrupt = 0;

	// No thread id until it is created
	m_hThread	= INVALID_HANDLE_VALUE;
	m_bIsPoolThread	= false;

#ifdef	_WIN32
	m_nThreadId = 0;
#else
	m_pThreadEvent = new REFThreadEvent(false);
#endif
}

/**
 * Cleanup thread static resources.
 */
void REFThread::StaticCleanup()
{
	if (s_bInitialized)
	{
		SpinLockHolder holder(&s_syncLock);
		if (s_bInitialized)
		{
			// initializes threading module...

#ifdef	_WIN32
			if (s_dwTlsSlot != 0)
			{
				// release TLS key.
				::TlsFree(s_dwTlsSlot);
				s_dwTlsSlot = 0;
			}
#else
			// TODO: Maybe we should send abort signals to all threads still alive, And wait for shutdown.

			// uninstall the thread alert signal handler.
			::sigaction(SIG_ALERT, NULL, NULL);

			// release TLS key.
			if (s_hTlsSlot != (pthread_key_t) INVALID_HANDLE_VALUE)
			{
				::pthread_key_delete(s_hTlsSlot);
				s_hTlsSlot = (pthread_key_t) INVALID_HANDLE_VALUE;
			}
#endif
			s_bInitialized = false;
		}
	}
}

/*
 * This is the backup startup routine.
 * It's keeps blank, will be invokes when thread starting up and no external delegate specified. We can overrides
 * this method in derived classes.
 */
void REFThread::Run()
{
	// Nothing to do.
}

// This method to handle interruption internally.
void REFThread::HandleThreadInterrupt()
{
	if (m_lUserInterrupt & TI_Abort)
	{
		// If the thread is waiting for abort, and the thread is interrupted,
		// we can start aborting.
		HandleThreadAbort();
	}
	if (m_lUserInterrupt & TI_Interrupt)
	{
		Interlocked::And(&m_lState, ~(TS_Interrupted | TS_Interruptible));
		Interlocked::And(&m_lUserInterrupt, ~TI_Interrupt);

		// I hate C++ exception, but we have to do it.
		ThrowAlertable(false);
	}
}

// This method to handle abortion internally.
void REFThread::HandleThreadAbort()
{
	Interlocked::And(&m_lState, ~(TS_Interrupted | TS_Interruptible));
	// We are going to abort.  Abort satisfies Interrupt() requirement.
	Interlocked::Exchange(&m_lUserInterrupt, 0);

	SetAbortInitiated();

	// I hate C++ exception, but we have to do it.
	ThrowAlertable(true);
}

// The helper methods called by BFX::REFThread::HandleThreadAbort() to sets thread state.
void REFThread::SetAbortInitiated()
{
	Interlocked::Or(&m_lState, TS_AbortInitiated);
	// The following should be factored better, but I'm looking for a minimal V1 change.
	Interlocked::Exchange(&m_lUserInterrupt, 0);
}

/*
 * Provides a appropriate(interruptible) wait functionality called by framework.
 * @see BFX::SyncBlock::Wait()
 */
bool REFThread::DoAppropriateWait(SyncBlock* pSyncBlk, INT32 nMilliseconds)
{
	BFX_ASSERT(nMilliseconds >= 0 || nMilliseconds == (INT32)INFINITE);
	bool bAlertable = true;

	if (bAlertable)
	{
		// NOTE: If thread abort is prevented???

		// A word about ordering for Interrupt.  If someone tries to interrupt a thread
		// that's in the interruptible state, we queue an APC.  But if they try to interrupt
		// a thread that's not in the interruptible state, we just record that fact.  So
		// we have to set TS_Interruptible before we test to see whether someone wants to
		// interrupt us or else we have a race condition that causes us to skip the APC.
		Interlocked::Or(&m_lState, TS_Interruptible);
		HandleThreadInterrupt();

		// Safe to clear the interrupted state, no signal could have fired since we
		// reset m_UserInterrupt (which inhibits our signal callback from doing
		// anything).
		Interlocked::And(&m_lState, ~TS_Interrupted);
	}

	ThreadStateHolder tsh(bAlertable, REFThread::TS_Interruptible | REFThread::TS_Interrupted);

	// Wait for the event to become signaled or timeout.
	bool bWaitResult = pSyncBlk->WaitNoAlert(nMilliseconds);

	if (bWaitResult)
	{
		// we've been interrupted
		BFX_ASSERT (bAlertable);
		if (m_lState & TS_Interrupted)
			HandleThreadInterrupt();
	}

	return bWaitResult;
}

/*
 * Throws alertable exceptions.
 * @param bAbort True indicates throw abort exception, otherwise interrupt exception.
 */
void REFThread::ThrowAlertable(bool bAbort)
{
	BFX_DBGTRACE("ThrowAlertable");
	if (bAbort)
		throw ThreadAbortException();
	else
		throw ThreadInterruptedException();
}

// The intermediate thread procedure, uses to suit for different systems. It's also initializes TLS data;
// autorelease pool...etc.
#ifdef	_WIN32
UINT32 __stdcall REFThread::IntermediateThreadProc(void* pContext)
#else
void* REFThread::IntermediateThreadProc(void* pContext)
#endif
{
	BFX_REQUIRE1(pContext != NULL, "Expect for an pointer of REFThread object.");

	REFThread* pThread = reinterpret_cast<REFThread*>(pContext);
	BFX_ASSERT(pThread->IsValid());

	// reset any unstarted bits on the thread object
	Interlocked::And(&pThread->m_lState, ~TS_Unstarted);

	// Wait for thread object ready.
	while (pThread->m_hThread == INVALID_HANDLE_VALUE)
	{
		REFThread::SpinWait();
	}

	// put our thread object into TLS. so we can uses BFX::REFThread::GetCurrentThread() method to access thread object later.
#ifdef	_WIN32
	::TlsSetValue(s_dwTlsSlot, pThread);
#else
	::pthread_setspecific(s_hTlsSlot, pThread);
#endif

	try
	{
		// Install a new autorelease pool for this thread.
		// NOTE: Each thread require least 1 autorelease pool in its entire lifecycle.
		REF<REFAutoreleasePool> pPool = new REFAutoreleasePool();

		// Call the startup routine
		REFThreadCallback* pCallback = pThread->m_pStart;
		if (pCallback != NULL)
			pCallback->Run();
		else
			pThread->Run();
	}
	catch (const Exception& e)
	{
		BFX_DBGWARN_("REFThread")("unhandled exception:(%s) occurred.", e.GetExceptionName());
	}
	catch (...)
	{
		BFX_DBGWARN_("REFThread")("unspecified exception occurred.");
	}
#ifndef	_WIN32
	// wakes up all waiting threads.
	pThread->m_pThreadEvent->Set();
#endif

	// cleanup before exit.
	pThread->m_hThread = INVALID_HANDLE_VALUE;
	Interlocked::Or(&pThread->m_lState, TS_Dead);

	if (pThread->m_bIsPoolThread)
	{
		// NOTE: pooled-thread need to release themselves.
		BFX_DBGTRACE("release pooled thread.");
		pThread->Release();
	}

#ifdef	_WIN32
	return 0;
#else
	return NULL;
#endif
}

// This is the callback from the OS, when we call ::pthread_kill send a signal to interrupt a waiting thread.
// The callback occurs on the thread we wish to interrupt.  It is a STATIC method.
#ifdef	_WIN32
void __stdcall REFThread::UserInterruptAPC(ULONG_PTR dwData)
{
	BFX_ASSERT(APC_ALERT == dwData);
#else
void REFThread::SignalInterruptProc(int nSignal)
{
	BFX_ASSERT(SIG_ALERT == nSignal);
#endif

    // Get our thread object from TLS.
	REFThread *pCurThread = GetCurrentThread();
	if (pCurThread)
	{
		// We should only take action if an interrupt is currently being requested
		// (our synchronization does not guarantee that we won't fire spuriously).
		// It's safe to check the m_lUserInterrupt field and then set TS_Interrupted
		// in a non-atomic fashion because m_lUserInterrupt is only cleared in this thread's context
		// (though it may be set from any context).
		if (pCurThread->m_lUserInterrupt)
		{
			// Set bit to indicate this routine was called (as opposed to other generic APCs).
			Interlocked::Or(&pCurThread->m_lState, TS_Interrupted);
		}
	}
}

} /* namespace BFX */
